#include <string.h>
#include <pthread.h>
#include <stdio.h>
#include <iostream>
#include <deque>

using namespace std;

#define MAX_QUEUE_SIZE   20
#define MAX_PRODUCER      3
#define MAX_CONSUMER      3
#define MAX_VALUES      1000000
#define PRO_TIMEOUT      0 //50
#define CON_TIMEOUT     0 //100

static void __nsleep(const struct timespec *req, struct timespec *rem) {
    struct timespec tempRem;
    if (nanosleep(req, rem) == -1)
        __nsleep(rem, &tempRem);

}

static void msSleep(unsigned long ms) {
    if (ms == 0)
        return;

    struct timespec req = {0};
    struct timespec rem = {0};

    time_t sec = (int) (ms / 1000);
    ms = ms - (sec * 1000);
    req.tv_sec = sec;
    req.tv_nsec = ms * 1000000L;

    __nsleep(&req, &rem);
}

class SharedData {
private:
    deque <int> list;
public:
    pthread_mutex_t lock;
    pthread_cond_t condNotFull;
    pthread_cond_t condNotEmpty;
    int messageCount;

    SharedData() {
        messageCount = 0;
    }

    bool dataAvailable() {
        bool rv = false;
        pthread_mutex_lock(&lock);
        rv = !list.empty();
        pthread_mutex_unlock(&lock);
        return rv;
    }

    void doConsume(int consumerId) {
        pthread_mutex_lock(&lock);
        while (list.empty() && this->messageCount < MAX_VALUES)pthread_cond_wait(&condNotEmpty, &lock);

        if (!list.empty()) {
            int currentMessage = list.front();
            list.pop_front();

            cout << "Message #" << currentMessage << "/" << MAX_VALUES << " consumed by #" << consumerId;
            cout << " Queuesize:" << list.size() << "/" << MAX_QUEUE_SIZE << endl;
        }
        if (list.size() < MAX_QUEUE_SIZE)
            pthread_cond_signal(&condNotFull);

        //Free all waiting Consumer for shutdown
        if (list.empty() && this->messageCount >= MAX_VALUES)
            pthread_cond_broadcast(&condNotEmpty);
        
        pthread_mutex_unlock(&lock);
    }

    void doProduce(int producerId) {
        pthread_mutex_lock(&lock);

        while (list.size() == MAX_QUEUE_SIZE && this->messageCount < MAX_VALUES)pthread_cond_wait(&condNotFull, &lock);
        if (this->messageCount < MAX_VALUES) {
            this->messageCount += 1;
            list.push_back(messageCount);

            cout << "Message #" << messageCount << "/" << MAX_VALUES << " produced by #" << producerId;
            cout << " Queuesize:" << list.size() << "/" << MAX_QUEUE_SIZE << endl;
        }
        if (list.size() == 1)
            pthread_cond_signal(&condNotEmpty);
        
        //Free all waiting Producer for shutdown
        if (this->messageCount >= MAX_VALUES)
            pthread_cond_broadcast(&condNotFull);
        
        pthread_mutex_unlock(&lock);
    }
};

struct ThreadData {
    SharedData* data;
    int index;
};

static void* producer(void* tData) {
    ThreadData* threadData = (ThreadData*) tData;
    SharedData* data = threadData->data;
    for (; data->messageCount < MAX_VALUES;) {
        msSleep(PRO_TIMEOUT);
        data->doProduce(threadData->index);
    }
    cout << "Exit Producer#" << threadData->index << endl;
}

static void* consumer(void* tData) {
    ThreadData* threadData = (ThreadData*) tData;
    SharedData* data = threadData->data;
    for (; (data->messageCount < MAX_VALUES || data->dataAvailable());) {
        msSleep(CON_TIMEOUT);
        data->doConsume(threadData->index);
    }
    cout << "Exit Consumer#" << threadData->index << endl;
}

int main() {
    struct SharedData shared;
    long error;
    pthread_t tproducer[MAX_PRODUCER];
    pthread_t tconsumer[MAX_CONSUMER];

    pthread_mutex_init(&shared.lock, NULL);
    pthread_cond_init(&shared.condNotFull, NULL);
    pthread_cond_init(&shared.condNotEmpty, NULL);



    for (int i = 0; i < MAX_PRODUCER; i++) {

        ThreadData* threadData = new ThreadData();
        threadData->data = &shared;
        threadData->index = i + 1;
        error = pthread_create(&tproducer[i], NULL, producer, threadData);
        if (error != 0) {
            fprintf(stderr, "pthread_create producer: %s\n", strerror(error));
            return 1;
        }
    }
    for (int i = 0; i < MAX_CONSUMER; i++) {

        ThreadData* threadData = new ThreadData();
        threadData->data = &shared;
        threadData->index = i + 1;
        error = pthread_create(&tconsumer[i], NULL, consumer, threadData);
        if (error != 0) {
            fprintf(stderr, "pthread_create consumer: %s\n", strerror(error));
            return 1;
        }
    }



    for (int i = 0; i < MAX_PRODUCER; i++) {
        error = pthread_join(tproducer[i], NULL);
        if (error != 0) {
            fprintf(stderr, "pthread_join producer: %s\n", strerror(error));
            return 1;
        }
    }
    for (int i = 0; i < MAX_CONSUMER; i++) {
        error = pthread_join(tconsumer[i], NULL);
        if (error != 0) {
            fprintf(stderr, "pthread_join consumer: %s\n", strerror(error));
            return 1;
        }
    }

    pthread_cond_destroy(&shared.condNotFull);
    pthread_cond_destroy(&shared.condNotEmpty);
    pthread_mutex_destroy(&shared.lock);
    cout << "Application exiting";
    return 0;
}

